package com.suxg.hbase.utils;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.springframework.util.ObjectUtils;

/**
 * Hbase 工具转换
 */
public class HBaseUtilsFilters {

	private static Connection connection;

	static {
		// localhost
		Configuration configuration = HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.property.clientPort", "2181");
		// 如果是集群 则主机名用逗号分隔
		configuration.set("hbase.zookeeper.quorum", "localhost:2181");
		configuration.set("hbase.nodeParent", "localhost:2181");
		// aliyun
//		Configuration configuration = HBaseConfiguration.create();
//		configuration.set("hbase.zookeeper.property.clientPort", "2181");
//		// 如果是集群 则主机名用逗号分隔
//		configuration.set("hbase.zookeeper.quorum", "101.200.212.148:2181");
//		configuration.set("hbase.nodeParent", "101.200.212.148:2181");

		try {
			connection = ConnectionFactory.createConnection(configuration);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 根据 rowKey 获取指定行的数据(Hbase shell: get)
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param famliy
	 * @param object
	 * @return
	 * @throws Exception
	 */
	public static Object getRow(String tableName, String rowKey, String famliy, Object object) throws Exception {
		Table table = connection.getTable(TableName.valueOf(tableName));
		ConverHbaseMapToObject(rowKey, famliy, object, table);
		return object;
	}

	/**
	 * 获取指定行指定列 (cell) 的最新版本的数据
	 *
	 * @param tableName    表名
	 * @param rowKey       唯一标识
	 * @param columnFamily 列族
	 * @param qualifier    列标识
	 */
	public static String getCell(String tableName, String rowKey, String columnFamily, String qualifier) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Get get = new Get(Bytes.toBytes(rowKey));
			if (!get.isCheckExistenceOnly()) {
				get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
				Result result = table.get(get);
				byte[] resultValue = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
				return Bytes.toString(resultValue);
			} else {
				return null;
			}

		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 查询后转换对象
	 * 
	 * @param rowKey
	 * @param family
	 * @param object
	 * @param table
	 * @throws IOException
	 * @throws IllegalAccessException
	 * @throws InvocationTargetException
	 */
	public static void ConverHbaseMapToObject(String rowKey, String family, Object object, Table table)
			throws IOException, IllegalAccessException, InvocationTargetException {
		Get get = new Get(Bytes.toBytes(rowKey));
		Result resultCell = table.get(get);
		Map<byte[], byte[]> familyMap = resultCell.getFamilyMap(Bytes.toBytes(family));
		Map<String, String> maps = new HashMap<>();
		for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
			System.out.println(Bytes.toString(entry.getKey()) + "=====>" + Bytes.toString((entry.getValue())));
			if (ObjectUtils.isEmpty(entry.getValue())) {
				maps.put(Bytes.toString(entry.getKey()), "");
			} else {
				maps.put(Bytes.toString(entry.getKey()), Bytes.toString(entry.getValue()));
			}
		}
		BeanUtils.populate(object, maps);
	}

	/**
	 * 检索表中指定数据
	 *
	 * @param tableName  表名
	 * @param filterList 过滤器
	 */

	public static ResultScanner getScanner(String tableName, FilterList filterList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.setFilter(filterList);
			return table.getScanner(scan);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 检索表中指定数据
	 *
	 * @param tableName   表名
	 * @param startRowKey 起始 RowKey
	 * @param endRowKey   终止 RowKey
	 * @param filterList  过滤器
	 */

	public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey,
			FilterList filterList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.setStartRow(Bytes.toBytes(startRowKey));
			scan.setStopRow(Bytes.toBytes(endRowKey));
			scan.setFilter(filterList);
			return table.getScanner(scan);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 插入数据
	 *
	 * @param tableName        表名
	 * @param rowKey           唯一标识
	 * @param columnFamilyName 列族名
	 * @param pairList         列标识和值的集合
	 */
	public static boolean putRow(String tableName, String rowKey, String columnFamilyName,
			List<Pair<String, String>> pairList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
//            pairList.forEach(pair -> put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(pair.getKey()), Bytes.toBytes(pair.getValue())));

			for (Pair<String, String> pair : pairList) {
				put.addColumn(Bytes.toBytes(columnFamilyName), pair.getFirst().getBytes(), pair.getSecond().getBytes());
			}
			table.put(put);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;

	}

	
	/**
	 * 全表scan 
	 * @param <T>
	 * @param scan 
	 * @param table
	 * @param object 返回的实体对象
	 * @return
	 */
	public static <T> List<T> ScannerList(Scan scan,Table table,Class<T> object) {

//		Table table = null;
		List<T> rs = new ArrayList<>();
		ResultScanner scanner = null;
		try {
//			table = connection.getTable(TableName.valueOf(tableName));
//			Scan scan = new Scan();
			// 进行全表扫描加开始，结束行，优化查询效率
//            scan.setStartRow(null);
//            scan.setStopRow(null);

//			Get get = new Get(Bytes.toBytes("rowKey1"));
//			Result resultCell = table.get(get);
//			Map<byte[], byte[]> familyMap = resultCell.getFamilyMap(Bytes.toBytes("student"));
//			for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
//				System.out.println(Bytes.toString(entry.getKey()));
//			}

			int caching = scan.getCaching();
			// 如果caching未设置(默认是1)，将默认配置成5000
			if (caching == 1) {
				scan.setCaching(5000);
			}
			scanner = table.getScanner(scan);
//			int rowNum = 0;
			Map<String, String> maps = new HashMap<>();
			T nweno = null;
			for (Result result : scanner) {
//				rs.add(mapper.mapRow(result, rowNum++, null));
				Iterator<KeyValue> it = result.list().iterator();
				while (it.hasNext()) {
					KeyValue kv = it.next();
//					CellUtil.getQualifierArray
					System.out.println("@@@@@@@@key@@@@key=" + Bytes.toString(kv.getQualifier()) + ", value=" + Bytes.toString(kv.getValue()));
					if (ObjectUtils.isEmpty(kv.getValue())) {
						maps.put(Bytes.toString(kv.getQualifier()), "");
					} else {
						maps.put(Bytes.toString(kv.getQualifier()), Bytes.toString(kv.getValue()));
					}
					nweno=object.newInstance();
				}
				BeanUtils.populate(nweno, maps);
				rs.add(nweno);
			
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			scanner.close();
		}
		return rs;
	}

	
	/**
	 * 检索全表
	 * 
	 * @param <T>
	 *
	 * @param tableName 表名
	 */
	public static <T> List<T> ScannerList(String tableName, Class<T> object) {

		Table table = null;
		List<T> rs = new ArrayList<>();
		ResultScanner scanner = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			// 进行全表扫描加开始，结束行，优化查询效率
//            scan.setStartRow(null);
//            scan.setStopRow(null);

//			Get get = new Get(Bytes.toBytes("rowKey1"));
//			Result resultCell = table.get(get);
//			Map<byte[], byte[]> familyMap = resultCell.getFamilyMap(Bytes.toBytes("student"));
//			for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
//				System.out.println(Bytes.toString(entry.getKey()));
//			}

			int caching = scan.getCaching();
			// 如果caching未设置(默认是1)，将默认配置成5000
			if (caching == 1) {
				scan.setCaching(5000);
			}
			scanner = table.getScanner(scan);
//			int rowNum = 0;
			Map<String, String> maps = new HashMap<>();
			T nweno = null;
			for (Result result : scanner) {
//				rs.add(mapper.mapRow(result, rowNum++, null));
				Iterator<KeyValue> it = result.list().iterator();
				while (it.hasNext()) {
					KeyValue kv = it.next();
//					CellUtil.getQualifierArray
					System.out.println("@@@@@@@@key@@@@key=" + Bytes.toString(kv.getQualifier()) + ", value=" + Bytes.toString(kv.getValue()));
					if (ObjectUtils.isEmpty(kv.getValue())) {
						maps.put(Bytes.toString(kv.getQualifier()), "");
					} else {
						maps.put(Bytes.toString(kv.getQualifier()), Bytes.toString(kv.getValue()));
					}
					nweno=object.newInstance();
				}
				BeanUtils.populate(nweno, maps);
				rs.add(nweno);
			
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			scanner.close();
		}
		return rs;
	}

}
